package com.tjhk.common.utils.thread.async;

import com.tjhk.common.utils.thread.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.Consumer;

/**
 * @author ：little grape
 * @description：TODO
 * @date ：2023/7/16 18:35
 *
 * maxRecordSize 最大批量处理记录数
 * keepAliveTime ThreadPoolExecutor keepAliveTime
 * RejectPolicy 线程异常策略
 * taskMethod 批处理任务方法
 *
 *  *   //构建工具类
 *  *         AsyncUtil<String> asyncUtil = new AsyncUtil(2,4,10,"test");
 *  **
 *  *         //注册执行批处理任务方法
 *  *         asyncUtil.registerMethod(list -> {
 *  * //            System.out.println("批处理方法将处理记录数："+list.size()+"=="+list.stream().collect(Collectors.joining(",")));
 *  *             ThreadUtil.sleep(1000);
 *  *         });
 *  *         //执行
 *  *         asyncUtil.start();
 *  *
 *  *         //模拟业务场景  单条处理
 *  *         new Thread(()->{
 *  *             for (int i = 0; i < 10000; i++) {
 *  *                 asyncUtil.add("结果"+i);
 *  *                 ThreadUtil.sleep(1);
 *  *             }
 *  *         }).run();
 *  *
 *  *         ThreadUtil.sleep(30000);
 *  *
 *
 *  建议使用BathConsumeUtil
 */
@Deprecated
public class AsyncUtil<T> {

    private static final Logger log = LoggerFactory.getLogger(AsyncUtil.class);

    private int maxRecordSize;
    private int keepAliveTime = 1;
    private TimeUnit timeUnit = TimeUnit.SECONDS;
    private ThreadPoolExecutor executor;
    private int poolSize;
    private int maximumPoolSize;
    private RejectPolicy rejectPolicy = RejectPolicy.DISCARD;
    private ConcurrentLinkedQueue<T> queue = new ConcurrentLinkedQueue<>();
    private Consumer<List<T>> taskMethod;
    private String taskName="task";
    /**等待任务时间**/
    private long maxWaitTime = 0;

    public AsyncUtil(int maxRecordSize) {
        this.maxRecordSize = maxRecordSize;
        this.poolSize = Runtime.getRuntime().availableProcessors();
        this.maximumPoolSize = this.poolSize * 2;
        this.executor = new ThreadPoolExtUtil(this.poolSize,this.maximumPoolSize,this.keepAliveTime,this.timeUnit,
                new LinkedBlockingQueue(64),
                new NamedThreadFactory("Async-"+ this.taskName+"-",true),
                new RejectedPolicyWithReport(this.rejectPolicy));
    }

    public AsyncUtil(int poolSize,int maximumPoolSize,int maxRecordSize, String taskName) {
        this.maxRecordSize = maxRecordSize;
        this.poolSize = poolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.taskName = taskName;
        this.executor = new ThreadPoolExtUtil(this.poolSize,this.maximumPoolSize,this.keepAliveTime,this.timeUnit,
                new LinkedBlockingQueue(64),
                new NamedThreadFactory("Async-"+ this.taskName+"-",true),
                new RejectedPolicyWithReport(this.rejectPolicy));
    }

    public AsyncUtil(int poolSize,int maximumPoolSize,int maxRecordSize, int keepAliveTime, String taskName, TimeUnit timeUnit,RejectPolicy rejectPolicy) {
        this.maxRecordSize = maxRecordSize;
        this.keepAliveTime = keepAliveTime;
        this.poolSize = poolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.taskName = taskName;
        this.timeUnit = timeUnit;
        this.rejectPolicy = rejectPolicy;
        this.executor = new ThreadPoolExtUtil(this.poolSize,this.maximumPoolSize,this.keepAliveTime,this.timeUnit,
                new LinkedBlockingQueue(64),
                new NamedThreadFactory("Async-"+ this.taskName+"-",true),
                new RejectedPolicyWithReport(this.rejectPolicy));
    }

    public void start(){
        check();
        log.info("批处理任务开始,最小线程数:{}========",this.poolSize);
        for (int i = 0; i < poolSize; i++) {
            this.executor.submit(()->{
                run();
            });
        }
    }

    private void run(){
        LinkedList<T> array = new LinkedList<>();
        long start = 0;
        for (; ; ) {
            T temp;
            int i = 0;
            while ((temp = queue.poll()) != null) {
                array.add(temp); i++;
                if (i >= maxRecordSize)
                    break;
            }

            if (i > 0) {
                start = System.currentTimeMillis();
                try {
                    log.info("task:{}==执行批处理任务，共处理{}条记录",Thread.currentThread().getName(),i);
                    List<T> list = new ArrayList<>();
                    for (int j = 0; j < i; j++) {
                        list.add(array.removeFirst());
                    }
                    runTask(list);
                } catch (Exception e) {
                    log.error("task:{}==执行批处理任务失败,e={},size={}",Thread.currentThread().getName(),e,i);
                }
                long time = System.currentTimeMillis() - start;
                if (time > 1000L)
                    log.warn("执行批处理任务耗时较长,共耗时:{}ms,共{}条记录", time, i);
            }

            if (i < maxRecordSize && queue.size() > 0){
                log.warn("task:{}==待处理记录数:{},未到达maxRecordSize将阻塞任务maxWaitTime:{}ms,目前活动线程数:{},建议减少线程数或调高maxRecordSize",Thread.currentThread().getName(),queue.size(),maxWaitTime,executor.getActiveCount());
                if (maxWaitTime < 3000L)
                    maxWaitTime+=200L;
                ThreadUtil.sleep(maxWaitTime);
            }

            if (queue.size() > maxRecordSize * 100){
                log.warn("task:{}==批处理队列繁忙, 待处理记录数:{},等待队列数：{},建议提高maxRecordSize或者提高核心数poolSize",Thread.currentThread().getName(),queue.size(),executor.getQueue().size());
                if (maxWaitTime > 0)
                    maxWaitTime = 0;
            }
        }
    }

    public void add(T t){
        queue.offer(t);
    }

    private void check(){
        if (taskMethod == null)
            throw new RuntimeException("未定义执行批量任务方法,请设置registerMethod方法！");
    }

    private void runTask(List<T> list){
        taskMethod.accept(list);
    }

    public void registerMethod(Consumer<List<T>> taskMethod){
        this.taskMethod = taskMethod;
    }

}
